package cn._51doit.flink.day02;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class CategoryNameFunction extends RichMapFunction<String, Tuple5<String, String, Integer, String, Double>> {

    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://node-1.51doit.cn:3306/doit28?characterEncoding=utf-8", "root", "123456");
    }

    @Override
    public Tuple5<String, String, Integer, String, Double> map(String line) throws Exception {
        String[] fields = line.split(",");
        String uid = fields[0];
        String oid = fields[1];
        int cid = Integer.parseInt(fields[2]);
        double money = Double.parseDouble(fields[3]);
        PreparedStatement preparedStatement = connection.prepareStatement("select name from tb_category where id = ?");
        preparedStatement.setInt(1, cid);
        ResultSet resultSet = preparedStatement.executeQuery();
        String name = null;
        if (resultSet.next()) {
            name = resultSet.getString(1);
        } else {
           name = "未知";
        }
        resultSet.close();
        preparedStatement.close();

        return Tuple5.of(uid, oid, cid, name, money);
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
